package com.thoughtful.remoting.transport.netty.codec;

import java.util.concurrent.atomic.AtomicInteger;

import com.thoughtful.compress.Compress;
import com.thoughtful.remoting.constants.RpcConstants;
import com.thoughtful.remoting.dto.RpcMessage;
import com.thoughtful.serialize.Serializer;
import com.thougtful.enums.CompressTypeEnum;
import com.thougtful.enums.SerializationTypeEnum;
import com.thougtful.extension.ExtensionLoader;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import lombok.extern.slf4j.Slf4j;

/**
 * <p>
 * 自定义协议编码器
 * <pre>
 *   0     1     2     3     4        5     6     7     8         9          10      11     12  13  14   15 16
 *   +-----+-----+-----+-----+--------+----+----+----+------+-----------+-------+----- --+-----+-----+-------+
 *   |   magic   code        |version | full length         | messageType| codec|compress|    RequestId       |
 *   +-----------------------+--------+---------------------+-----------+-----------+-----------+------------+
 *   |                                                                                                       |
 *   |                                         body                                                          |
 *   |                                                                                                       |
 *   |                                        ... ...                                                        |
 *   +-------------------------------------------------------------------------------------------------------+
 * 4B magic code (魔法数)    1B version (版本)       4B full length (消息长度)    1B messageType (消息类型)
 * 1B codec (序列化类型)     1B compress (压缩类型)   4B  requestId (请求的Id)
 * body (object类型数据)
 * </pre>
 *
 * @author yushun
 * @createTime 2022/10/18 - 10:31
 **/
@Slf4j
public class RpcMessageEncoder extends MessageToByteEncoder<RpcMessage> {

    private static final AtomicInteger ATOMIC_INTEGER = new AtomicInteger(0);

    @Override
    protected void encode(ChannelHandlerContext ctx, RpcMessage msg, ByteBuf out) throws Exception {
        try {
            out.writeBytes(RpcConstants.MAGIC_NUMBER);
            out.writeByte(RpcConstants.VERSION);
            // 空下 4 字节用于写入消息长度
            out.writerIndex(out.writerIndex() + 4);
            byte messageType = msg.getMessageType();
            out.writeByte(messageType);
            out.writeByte(msg.getCodec());
            out.writeByte(msg.getCompress());
            out.writeInt(ATOMIC_INTEGER.getAndUpdate(operand -> {
                if (operand == Integer.MAX_VALUE) {
                    return 1;
                }
                return operand + 1;
            }));

            byte[] bodyBytes = null;
            int fullLength = RpcConstants.HEAD_LENGTH;

            // 如果不是心跳包, 则消息整体的长度 = 消息头长度 + 消息体长度
            if (messageType != RpcConstants.HEARTBEAT_REQUEST_TYPE
                && messageType != RpcConstants.HEARTBEAT_RESPONSE_TYPE) {
                // 序列化 data 数据
                String codeType = SerializationTypeEnum.getName(msg.getCodec());
                log.info("codec type: [{}]", codeType);
                Serializer serializer = ExtensionLoader.getExtensionLoader(Serializer.class).getExtension(codeType);

                bodyBytes = serializer.serializer(msg.getData());

                // 压缩
                String compressType = CompressTypeEnum.getName(msg.getCompress());
                Compress compress = ExtensionLoader.getExtensionLoader(Compress.class).getExtension(compressType);

                bodyBytes = compress.compress(bodyBytes);
                fullLength += bodyBytes.length;
            }

            if (bodyBytes != null) {
                out.writeBytes(bodyBytes);
            }
            int writerIndex = out.writerIndex();
            out.writerIndex(writerIndex - fullLength + RpcConstants.MAGIC_NUMBER.length + 1);
            out.writeInt(fullLength);
            out.writerIndex(writerIndex);
        } catch (Exception e) {
            log.error("Encode request error!", e);
        }
    }

}
